package edu.maxflow.mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;

import edu.maxflow.Edge;
import edu.maxflow.Path;
import edu.maxflow.Vertex;
import edu.maxflow.mapreduce.util.Accumulator;
import edu.maxflow.mapreduce.util.MRConstants;

public class MaxFlowMapper extends Mapper<LongWritable, Vertex, LongWritable, Vertex> {
	Accumulator augmentedPathAccumulator = new Accumulator();

	public void map(LongWritable key, Vertex vertex, Context context) throws IOException, InterruptedException {
		// get configuration variables
		Configuration conf = context.getConfiguration();
		long targetVertexId = conf.getInt(MRConstants.TARGET_VERTEX_ID, MRConstants.TARGET_VERTEX_DEFAULT_VALUE);
		int round = conf.getInt(MRConstants.ROUND, MRConstants.ROUND_DEFAULT_VALUE);

		List<Edge> augmentedEdges = new ArrayList<Edge>();
		if (round > 1) {
			FSDataInputStream inputStream = null;

			try {
				FileSystem fileSystem = FileSystem.get(conf);
				org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(MRConstants.PATH_AUGMENTED_EDGE + (round - 1));
				inputStream = fileSystem.open(path);

				int size = inputStream.readInt();
				for (int i = 0; i < size; i++) {
					Edge edge = new Edge();
					edge.readFields(inputStream);

					augmentedEdges.add(edge);
				}
			} catch (Exception e) {
				System.err.println("File read exception occured");
			} finally {
				if (inputStream != null) {
					inputStream.close();
				}
			}
		}

		List<Path> sourceExcessPaths = vertex.getSourceExcessPaths();
		List<Path> targetExcessPaths = vertex.getTargetExcessPaths();
		List<Edge> adjacentEdges = vertex.getAdjacentEdges();

		Iterator<Path> sourceExcessPathIterator = sourceExcessPaths.iterator();
		Iterator<Path> targetExcessPathIterator = targetExcessPaths.iterator();
		Iterator<Edge> adjacentEdgeIterator = adjacentEdges.iterator();

		// update edge flows based on previous round's accepted edges. Also
		// remove saturated edges.
		if (augmentedEdges.size() > 0) {
			if (sourceExcessPathIterator != null) {
				while (sourceExcessPathIterator.hasNext()) {
					Path sourceExcessPath = sourceExcessPathIterator.next();

					outerloop: for (Edge edge : sourceExcessPath.getEdgeList()) {
						for (Edge augmentedEdge : augmentedEdges) {
							if (edge.equals(augmentedEdge)) {
								long currentFlow = edge.getFlow() + augmentedEdge.getFlow();

								if (currentFlow > edge.getCapacity()) {
									sourceExcessPathIterator.remove();
								} else {
									edge.setFlow(currentFlow);
								}

								break outerloop;
							}
						}
					}
				}
			}

			if (targetExcessPathIterator != null) {
				while (targetExcessPathIterator.hasNext()) {
					Path targetExcessPath = targetExcessPathIterator.next();

					outerloop: for (Edge edge : targetExcessPath.getEdgeList()) {
						for (Edge augmentedEdge : augmentedEdges) {
							if (edge.equals(augmentedEdge)) {
								long currentFlow = edge.getFlow() + augmentedEdge.getFlow();

								if (currentFlow > edge.getCapacity()) {
									targetExcessPathIterator.remove();
								} else {
									edge.setFlow(currentFlow);
								}

								break outerloop;
							}
						}
					}
				}
			}

			if (adjacentEdgeIterator != null) {
				while (adjacentEdgeIterator.hasNext()) {
					Edge edge = adjacentEdgeIterator.next();

					for (Edge augmentedEdge : augmentedEdges) {
						if (edge.equals(augmentedEdge)) {
							long currentFlow = edge.getFlow() + augmentedEdge.getFlow();

							if (currentFlow > edge.getCapacity()) {
								adjacentEdgeIterator.remove();
							} else {
								edge.setFlow(currentFlow);
							}

							break;
						}
					}
				}
			}
		}

		// accumulator
		if (sourceExcessPaths != null && targetExcessPaths != null) {
			for (Path sourcePath : sourceExcessPaths) {
				for (Path targetPath : targetExcessPaths) {
					if (augmentedPathAccumulator.accept(sourcePath, targetPath)) {
						Vertex vertexToEmit = new Vertex();
						vertexToEmit.setVertexId(targetVertexId);

						vertexToEmit.getSourceExcessPaths().add(sourcePath);
						vertexToEmit.getSourceExcessPaths().get(0).getEdgeList().addAll(targetPath.getEdgeList());

						context.write(new LongWritable(vertexToEmit.getVertexId()), vertexToEmit);
					}
				}
			}
		}

		// emit vertices based on source excess paths
		if (sourceExcessPaths != null && sourceExcessPaths.size() != 0 && adjacentEdges != null) {
			for (Edge edge : adjacentEdges) {
				if (edge.getFlow() < edge.getCapacity()) {
					for (Path sourceExcessPath : sourceExcessPaths) {
						if (!sourceExcessPath.containsEdge(edge)) {
							List<Edge> newEdgeList = new ArrayList<Edge>(sourceExcessPath.getEdgeList());
							newEdgeList.add(edge);

							Path newSourceExcessPath = new Path();
							newSourceExcessPath.setEdgeList(newEdgeList);

							Vertex vertexToEmit = new Vertex();
							vertexToEmit.getSourceExcessPaths().add(newSourceExcessPath);

							if (key.get() == edge.getFromVertexId()) {
								vertexToEmit.setVertexId(edge.getToVertexId());
							} else {
								vertexToEmit.setVertexId(edge.getFromVertexId());
							}

							context.write(new LongWritable(vertexToEmit.getVertexId()), vertexToEmit);
						}
					}
				}
			}
		}

		// emit vertices based on target excess paths
		if (targetExcessPaths != null && targetExcessPaths.size() != 0 && adjacentEdges != null) {
			for (Edge edge : adjacentEdges) {
				if (edge.getFlow() < edge.getCapacity()) {
					for (Path targetExcessPath : targetExcessPaths) {
						if (!targetExcessPath.containsEdge(edge)) {
							List<Edge> newEdgeList = new ArrayList<Edge>(targetExcessPath.getEdgeList());
							newEdgeList.add(edge);

							Path newTargetExcessPath = new Path();
							newTargetExcessPath.setEdgeList(newEdgeList);

							Vertex vertexToEmit = new Vertex();
							vertexToEmit.getTargetExcessPaths().add(newTargetExcessPath);

							if (key.get() == edge.getFromVertexId()) {
								vertexToEmit.setVertexId(edge.getToVertexId());
							} else {
								vertexToEmit.setVertexId(edge.getFromVertexId());
							}

							context.write(new LongWritable(vertexToEmit.getVertexId()), vertexToEmit);
						}
					}
				}
			}
		}

		vertex.setAdjacentEdges(adjacentEdges);
		vertex.setSourceExcessPaths(sourceExcessPaths);
		vertex.setTargetExcessPaths(targetExcessPaths);
		// emit the original vertex
		context.write(key, vertex);
	}
}
